package db.prob.operators.projectjoin;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import db.prob.io.LineParser;
import db.prob.io.PTuple;
import db.prob.operators.join.MapperJoin;

public class ReducerProjectJoin extends Reducer<Text, Text, Text, Text> {
	@Override
	protected void reduce(Text key, Iterable<Text> values,Context context) throws IOException, InterruptedException {
		if (key.toString().equals(MapperProjectJoin.PASS)) {
			for (Text value : values) {
				context.write(null, value);
			}
			return;
		}
		
		Configuration conf = context.getConfiguration();
		String table1 = conf.get(MapperProjectJoin.TABLE_1);
		String table2 = conf.get(MapperProjectJoin.TABLE_2);

		
		// First we do the projection.
		Map<String, Float> table1ProjProb = new HashMap<String, Float>();
		Map<String, Float> table2ProjProb = new HashMap<String, Float>();

		for (Text entry : values) {
			PTuple tuple = LineParser.createPTuple(entry.toString(), "\\t");
			if (tuple.getTableName().equals(table1)) {
				handleTable(table1ProjProb, tuple);
			} else {
				handleTable(table2ProjProb, tuple);
			}
		}
		
		// before we go on we have to do a 1 - prob.
		do1MinusProb(table1ProjProb);
		do1MinusProb(table2ProjProb);
		
		// now we do the join!
		String joinTableName = conf.get(MapperProjectJoin.RESULT_TABLE);
		
		Text newVal = new Text();

		// We don't need this attribute since it appears twice.
		int table2Attr = conf.getInt(MapperJoin.TABLE_2_ATTR, Integer.MAX_VALUE);
		
		// Doing the join
		for (Entry<String, Float> tuple1 : table1ProjProb.entrySet()) {
			String tuple1Key = tuple1.getKey();
			String[] tuple1Attr = tuple1Key.substring(1, tuple1Key.length()-1).split("\\W+");

			for (Entry<String, Float> tuple2 : table2ProjProb.entrySet()) {
				
				float newProb = tuple1.getValue() * tuple2.getValue();
				
				String tuple2Key = tuple2.getKey();
				String[] tuple2Attr = tuple2Key.substring(1, tuple2Key.length()-1).split("\\W+");
				// copy all except for the duplicated value.
				tuple2Attr[table2Attr] = "";
				
				String[] newAttr = new String[tuple1Attr.length + tuple2Attr.length];
				System.arraycopy(tuple1Attr, 0, newAttr, 0, tuple1Attr.length);
				System.arraycopy(tuple2Attr, 0, newAttr, tuple1Attr.length, tuple2Attr.length);
				PTuple newTuple = new PTuple(joinTableName, newProb, newAttr);
				newVal.set(newTuple.toString());
				context.write(null, newVal);
			}
		}
		
	}

	private void do1MinusProb(Map<String, Float> table1ProjProb) {
		for (String key : table1ProjProb.keySet()) {
			table1ProjProb.put(key, 1 - table1ProjProb.get(key));
		}
	}

	/**
	 * @param table1ProjProb
	 * @param tuple
	 */
	private void handleTable(Map<String, Float> table1ProjProb, PTuple tuple) {
		double newProb;
		String arrayKey = Arrays.toString(tuple.getAttr());
		if (table1ProjProb.containsKey(arrayKey)) {
			newProb = table1ProjProb.get(arrayKey) * (1 - tuple.getProbability());
		} else {
			newProb = 1 - tuple.getProbability(); 
		}
		table1ProjProb.put(arrayKey, new Float(newProb));
	}

}
